# %%
# 初始化
import pandas as pd

import reader
from common import *
from common import last_n_month

(year, month), _ = year_month()
year_month_list = last_n_month(year, month)

FP_GONGDAN_LIST = [
    f"data/0-工单/家客工单导出_潮州_{year}-{month:02}.zip"
    for year, month in year_month_list
]
FP_GONGDAN_DANGYUE = (
    f"data/0-工单/家客工单导出_潮州_{year}-{month:02}.zip"
)
FP_HUGAN = f"data/0-互感/{year}{month:02}_互感数据.zip"
FP_GUANXIAN = f"data/0-管线/{year}{month:02}_管线数据.zip"
FP_ONU_GGL = (
    f"data/2-宽带装移机/{year}{month:02}_新装ONU管控报表清单.csv"
)
FP_ZHIJIAN = f"data/2-宽带装移机/{year}{month:02}_智能质检清单.xlsx"
FP_JIHE_MINGXI = (
    f"data/3-电视装移机/{year}{month:02}电视开通地市稽核明细报表.xlsx"
)
FP_TV_PINGTAI = f"data/3-电视装移机/{year}{month:02}_电视平台数据.zip"
FP_TV_YIDI_PINGTAI = (
    f"data/3-电视装移机/{year}{month:02}_潮州异地账号.csv"
)
FP_KUANDAI_JIHE_PKL = (
    f"pickle/{year}{month:02}_宽带装移机工单地市稽核明细表.pkl"
)

check_files(
    FP_GONGDAN_LIST,
    FP_HUGAN,
    FP_GUANXIAN,
    FP_ONU_GGL,
    FP_ZHIJIAN,
    FP_JIHE_MINGXI,
    FP_TV_PINGTAI,
    # FP_TV_YUEBAO,
    FP_KUANDAI_JIHE_PKL,
)


# %%
# 读取工单
usecols = (
    "工单号",
    "宽带帐号",
    "电视账号",
    "CRM工单号",
    "CRM业务流水号",
    "操作类型",
    "接入方式",
    "是否施工标识",
    "产品名称",
    "派单日期",
    "结单时间",
    "工单状态",
    "资源类型",
    "标准地址",
    "产品业务属性",
    "首次第三方复核结果",
)
df = pd.concat(
    [
        reader.read_gongdan(f, usecols)
        for f in FP_GONGDAN_LIST
    ],
    ignore_index=True,
)
df_dy = reader.read_gongdan(FP_GONGDAN_DANGYUE, usecols)


# %%
# 筛选工单
# 操作类型 = 业务开通、业务移机
# 是否施工 = 普通开通施工
# 产品名称 = 宽带电视基础产品
# 工单状态 =  归档
type_mask = lambda df: (
    (df["操作类型"] == "业务拆除")
    | (
        (df["操作类型"] == "业务开通")
        & ((df["产品业务属性"] == "新增业务") | df["产品业务属性"].isna())
    )
    | ((df["操作类型"] == "业务移机") & (df["产品业务属性"] == "存量业务"))
)

other_mask = lambda df: (
    (
        (df["是否施工标识"].isna())
        | (df["是否施工标识"] == "普通开通施工")
        | (df["是否施工标识"] == "")
    )
    & (df["产品名称"].isin(["宽带电视基础产品"]))
    & (df["工单状态"] == "归档")
)
df = df[type_mask(df)]
df = df[other_mask(df)]
df_dy = df_dy[type_mask(df_dy)]
df_dy = df_dy[other_mask(df_dy)]

df_dy_cj = df_dy[df_dy["操作类型"] == "业务拆除"]
df_dy_kt = df_dy[df_dy["操作类型"] == "业务开通"]
df_dy_yj = df_dy[df_dy["操作类型"] == "业务移机"]
df_dy = df_dy[df_dy["操作类型"].isin(["业务开通", "业务移机"])]

# %%
# 90天同址同帐号先拆后装核查
df_tmp = df_dy_kt.merge(
    df.loc[
        df["操作类型"] == "业务拆除",
        [
            "电视账号",
            "标准地址",
            "派单时间",
        ],
    ],
    on=["电视账号", "标准地址"],
)

mask = df_dy["工单号"].isin(
    df_tmp[
        df_tmp["派单时间_x"] - df_tmp["派单时间_y"]
        < pd.Timedelta(days=90)
    ]["工单号"]
)
df_dy["90天同址同帐号先拆后装核查"] = S_PASSED
df_dy.loc[
    df_dy["操作类型"] != "业务开通", "90天同址同帐号先拆后装核查"
] = S_BYPASSED
df_dy.loc[mask, "90天同址同帐号先拆后装核查"] = S_NOT_PASSED
# %%
# 当月同帐号先装后拆核查
df_tmp = df_dy_kt.merge(
    df_dy_cj[["电视账号", "派单时间"]],
    on=["电视账号"],
)

mask = df_dy["工单号"].isin(
    df_tmp[df_tmp["派单时间_x"] < df_tmp["派单时间_y"]]["工单号"]
)

df_dy["当月同帐号先装后拆核查"] = S_PASSED
df_dy.loc[
    df_dy["操作类型"] != "业务开通", "当月同帐号先装后拆核查"
] = S_BYPASSED
df_dy.loc[mask, "当月同帐号先装后拆核查"] = S_NOT_PASSED

# %%
# 90天内同帐号先装后移核查
df_tmp = df_dy_yj.merge(
    df.loc[
        df["操作类型"] == "业务开通",
        [
            "电视账号",
            "派单时间",
        ],
    ],
    on=["电视账号"],
)

mask = df_dy["工单号"].isin(
    df_tmp[
        df_tmp["派单时间_x"] - df_tmp["派单时间_y"]
        < pd.Timedelta(days=90)
    ]["工单号"]
)

df_dy["90天内同帐号先装后移核查"] = S_PASSED
df_dy.loc[
    df_dy["操作类型"] != "业务移机", "90天内同帐号先装后移核查"
] = S_BYPASSED
df_dy.loc[mask, "90天内同帐号先装后移核查"] = S_NOT_PASSED
# %%
# 90内同帐号多次移机核查
df_tmp = df_dy_yj.merge(
    df.loc[
        df["操作类型"] == "业务移机",
        [
            "工单号",
            "电视账号",
            "派单时间",
        ],
    ],
    on=["电视账号"],
)

mask = df_dy["工单号"].isin(
    df_tmp[
        (df_tmp["工单号_x"] != df_tmp["工单号_y"])
        & (
            df_tmp["派单时间_x"] - df_tmp["派单时间_y"]
            < pd.Timedelta(days=90)
        )
    ]["工单号_x"]
)

df_dy["90内同帐号多次移机核查"] = S_PASSED
df_dy.loc[
    df_dy["操作类型"] != "业务移机", "90内同帐号多次移机核查"
] = S_BYPASSED
df_dy.loc[mask, "90内同帐号多次移机核查"] = S_NOT_PASSED
# %%
# 当月相同联系电话装机工单核查（不稽核）
df_dy["当月相同联系电话装机工单核查"] = S_BYPASSED

# %%
# 当月新增一址多户工单核查
df_dy["当月新增一址多户工单核查"] = S_BYPASSED

df_guanxian = reader.read_csv(
    FP_GUANXIAN,
    columns=["宽带号码", "七级地址ID"],
)
guanxian_err_addr = df_guanxian.groupby("七级地址ID").filter(
    lambda x: len(x) >= 5
)["宽带号码"]
df_dy["当月新增一址多户工单核查"] = S_PASSED
df_dy.loc[
    df_dy["宽带帐号"].isin(guanxian_err_addr), "当月新增一址多户工单核查"
] = S_NOT_PASSED

# %%
# 登录端口异常工单核查
df_dy["登录端口异常工单核查"] = S_PASSED

df_hugan = reader.read_csv(
    FP_HUGAN,
    columns=(
        "宽带号码",
        "ONU下联端口",
        "所属ONU设备名称",
        "错误类型",
    ),
)
df_hugan_err_port = df_hugan.groupby(
    ["ONU下联端口", "所属ONU设备名称"]
).filter(lambda x: len(x) >= 2)["宽带号码"]
df_dy.loc[
    df_dy["宽带帐号"].isin(df_hugan_err_port), "登录端口异常工单核查"
] = S_NOT_PASSED


# %%
# 当月登录端口与资源系统不一致核查
df_dy["当月登录端口与资源系统不一致核查"] = S_NOT_PASSED
df_hugan_ok_ziyuan = df_hugan[df_hugan["错误类型"] == "正确资源"][
    "宽带号码"
]
df_dy.loc[
    df_dy["宽带帐号"].isin(df_hugan_ok_ziyuan),
    "当月登录端口与资源系统不一致核查",
] = S_PASSED

# %%
# 电视账号未激活核查
df_tv_pingtai = reader.read_csv(
    FP_TV_PINGTAI,
    columns=(
        "电视号码",
        "电视状态",
        "最后一次登陆时间",
    ),
    dtype="string",
)
df_tv_pingtai.rename(
    columns={
        "电视号码": "USERCODE",
        "电视状态": "STATUS",
        "最后一次登陆时间": "UPDATETIME",
    },
    inplace=True,
)

df_tv_yidi_pingtai = pd.read_csv(
    FP_TV_YIDI_PINGTAI,
    usecols=(
        "USERCODE",
        "STATUS",
        "UPDATETIME",
    ),
    dtype="string",
)
df_tv_pingtai = pd.concat(
    [df_tv_pingtai, df_tv_yidi_pingtai],
    ignore_index=True,
)

tv_ok = df_tv_pingtai.loc[
    (
        df_tv_pingtai["STATUS"].isin(["1", "4"])
        & (df_tv_pingtai.UPDATETIME >= f"{year}{month:02}")
    ),
    "USERCODE",
]
df_dy["电视账号未激活核查"] = S_NOT_PASSED
df_dy.loc[df_dy["电视账号"].isin(tv_ok), "电视账号未激活核查"] = S_PASSED


# %%
# 服开工单规范性质检
df_znzj = pd.read_excel(
    FP_ZHIJIAN,
    usecols=("工单编号", "业务类型"),
    dtype="string",
    skiprows=1,
)
df_znzj = df_znzj[df_znzj["业务类型"] == "电视"]["工单编号"]
mask = df_dy["工单号"].isin(df_znzj) | (
    df_dy["首次第三方复核结果"] == S_NOT_PASSED
)
df_dy["服开工单规范性质检"] = S_PASSED
df_dy.loc[mask, "服开工单规范性质检"] = S_NOT_PASSED


# %%
# 弱光核查
dtype_dict = {
    "工单号": "string",
    "操作类型": "string",
    "产品类型": "string",
    "光功率是否达标": "string",
    "整改后光功率值": "float",
}
df_ggl = pd.read_csv(
    FP_ONU_GGL,
    usecols=dtype_dict.keys(),
    encoding="gbk",
    dtype=dtype_dict,
)

ggl_ok = df_ggl.loc[
    (df_ggl["产品类型"] == "互联网电视")
    & (
        (df_ggl["光功率是否达标"] == "是")
        | (df_ggl["整改后光功率值"] >= -25)
    ),
    "工单号",
]
df_dy["弱光核查"] = S_NOT_PASSED
df_dy.loc[
    df_dy["工单号"].isin(ggl_ok),
    "弱光核查",
] = S_PASSED
df_dy.loc[
    (df_dy["接入方式"] == "FTTB"),
    "弱光核查",
] = S_BYPASSED

# %%
# 宽带稽核是否通过核查
df_kuandai = pd.read_pickle(FP_KUANDAI_JIHE_PKL)
df_tmp = df_dy[["CRM业务流水号", "宽带帐号"]].merge(
    df_kuandai[["宽带帐号", "地市稽核结果"]],
    on="宽带帐号",
    how="left",
)
df_dy["宽带稽核是否通过核查"] = S_PASSED
mask = df_dy["CRM业务流水号"].isin(
    df_tmp[df_tmp["地市稽核结果"] == S_NOT_PASSED]["CRM业务流水号"]
)
df_dy.loc[mask, "宽带稽核是否通过核查"] = S_NOT_PASSED


# %%
CHECK_COLUMNS = [
    "90天同址同帐号先拆后装核查",
    "当月同帐号先装后拆核查",
    "90天内同帐号先装后移核查",
    "90内同帐号多次移机核查",
    "当月相同联系电话装机工单核查",
    "当月新增一址多户工单核查",
    "登录端口异常工单核查",
    "当月登录端口与资源系统不一致核查",
    "电视账号未激活核查",
    "弱光核查",
    "服开工单规范性质检",
    "宽带稽核是否通过核查",
]
df_dy["地市稽核结果"] = S_PASSED
not_passed = (df_dy[CHECK_COLUMNS] == S_NOT_PASSED).any(
    axis=1
)
df_dy.loc[not_passed, "地市稽核结果"] = S_NOT_PASSED

# %%
# 电视装移机地市稽核明细表
columns = [
    "地市编码",
    "地市名称",
    "账期",
    "工单号",
    "CRM单号",
    "宽带号码",
    "电视号码",
    "操作类型",
    "派单时间",
    "归档日期",
    "接入模式",
    "90天同址同帐号先拆后装核查",
    "当月同帐号先装后拆核查",
    "90天内同帐号先装后移核查",
    "90内同帐号多次移机核查",
    "当月相同联系电话装机工单核查",
    "当月新增一址多户工单核查",
    "登录端口异常工单核查",
    "当月登录端口与资源系统不一致核查",
    "电视账号未激活核查",
    "服开工单规范性质检",
    "弱光核查",
    "系统稽核结果",
]
df_sgs = pd.read_excel(FP_JIHE_MINGXI, usecols=columns)
df_sgs = df_sgs.merge(
    df_dy[["工单号", "宽带稽核是否通过核查", "地市稽核结果"]],
    on="工单号",
    how="left",
)

# %%
# 地市与系统稽核不一致原因
def explain(x):
    reasons = []
    for col in CHECK_COLUMNS:
        if (x[f"{col}_cz"] == S_NOT_PASSED) and (
            x[f"{col}_sgs"] == S_PASSED
        ):
            reasons.append(col)

    return ";".join(reasons)


df_tmp = df_sgs.merge(
    df_dy, on="工单号", how="left", suffixes=("_sgs", "_cz")
)
df_sgs["地市与系统稽核不一致原因"] = df_tmp.apply(explain, axis=1)
df_sgs.loc[~df_sgs.工单号.isin(df_dy.工单号), "市缺"] = S_YES
df_sgs.to_excel(
    f"report/3-电视装移机/{year}{month:02}_电视开通稽核地市铁通核对.xlsx",
    index=False,
)

# %%
# 电视开通工单稽核结果
df_dy = df_dy.merge(
    df_sgs[["工单号", "系统稽核结果"]], on="工单号", how="left"
)
df_dy.loc[~df_dy.工单号.isin(df_sgs.工单号), "省缺"] = S_YES
df_dy.to_excel(
    f"report/3-电视装移机/{year}{month:02}_电视装移机地市稽核明细表.xlsx",
    index=False,
)

# %%
# 完成
